package com.itgacl.magic4j.libcommon.component.delayQueue.manager;

import com.itgacl.magic4j.libcommon.component.delayQueue.bean.QueueData;
import com.itgacl.magic4j.libcommon.component.delayQueue.listener.MessageOutListener;
import com.itgacl.magic4j.libcommon.component.delayQueue.queue.DelayQueue;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;

/**
 * 延迟队列管理器
 *
 */
@Slf4j
@Component
public class DelayQueueManager implements DelayQueue {

    public static final long NOT_TIMESTAMP = -1L;

    private static final String CONCAT_SYMBOL="#";

    @Resource(name = "delayDb")
    private RocksDB db;

    @Autowired
    private MessageOutListener delayMessageOutListener;

    public int insert(String msgId, String message, Long timestamp) {
        try {
            String body = message.concat(CONCAT_SYMBOL).concat(timestamp.toString());
            db.put(msgId.getBytes(), body.getBytes());
        } catch (RocksDBException e) {
            log.error(e.getMessage(), e);
        }
        return 1;
    }

    public int update(String msgId, String message, Long timestamp) {
        try {
            String body = message.concat(CONCAT_SYMBOL).concat(timestamp.toString());
            db.put(msgId.getBytes(), body.getBytes());
        } catch (RocksDBException e) {
            log.error(e.getMessage(), e);
        }
        return 1;
    }

    public int delete(String msgId) {
        try {
            db.delete(msgId.getBytes());
        } catch (RocksDBException e) {
            log.error(e.getMessage(), e);
        }
        return 1;
    }

    public QueueData get(String msgId) {
        byte[] bytes = new byte[0];
        try {
            bytes = db.get(msgId.getBytes());
        } catch (RocksDBException e) {
            log.error(e.getMessage(), e);
        }
        QueueData queueData = new QueueData();
        queueData.setMsgId(msgId);
        String bodyMessage = new String(bytes);
        String[] elements = bodyMessage.split(CONCAT_SYMBOL);
        queueData.setMsgContent(elements[0]);
        long delayTime = Long.parseLong(elements[elements.length-1]);
        if (delayTime == NOT_TIMESTAMP) {
            queueData.setDelayTime(elements[1]);
        }else {
            Date tmpDate = new Date(delayTime);
            queueData.setDelayTime(formatDate(tmpDate));
        }
        return queueData;
    }

    public void sendDelayMessage() {
        RocksIterator iterator = null;
        try {
            iterator = db.newIterator();
            iterator.seekToFirst();
            for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
                byte[] body = iterator.value();
                byte[] key = iterator.key();
                String msgId = new String(key);
                String bodyMessage = new String(body);
                String[] elements = bodyMessage.split(CONCAT_SYMBOL);
                QueueData queueData = new QueueData();
                queueData.setMsgId(msgId);
                queueData.setMsgContent(elements[0]);
                if (elements.length == 2) {
                    long time = System.currentTimeMillis();
                    long delayTime = Long.parseLong(elements[elements.length-1]);
                    if (delayTime == NOT_TIMESTAMP) {
                        queueData.setDelayTime(elements[1]);
                        output(queueData);
                        continue;
                    }
                    if (time < delayTime) {
                        continue;
                    }
                    Date tmpDate = new Date(delayTime);
                    queueData.setDelayTime(formatDate(tmpDate));
                }
                log.info("DelayQueueManager sendDelayMessage，queueData：{}",queueData);
                output(queueData);
            }

        }catch (Exception e){
            log.error(e.getMessage(),e);
        } finally {
            if (iterator != null) {
                iterator.close();
            }
        }
    }

    /**
     * 发送指定消息ID的延迟消息
     */
    public void sendDelayMessage(String msgId){
        QueueData queueData = get(msgId);
        output(queueData);
    }

    public List<QueueData> getAll(){
        List<QueueData> queueDataList = new ArrayList<>();
        RocksIterator iterator = null;
        try {
            iterator = db.newIterator();
            iterator.seekToFirst();
            for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
                byte[] body = iterator.value();
                byte[] key = iterator.key();
                String msgId = new String(key);
                String bodyMessage = new String(body);
                String[] elements = bodyMessage.split(CONCAT_SYMBOL);
                QueueData queueData = new QueueData();
                queueData.setMsgId(msgId);
                queueData.setMsgContent(elements[0]);
                if (elements.length == 2) {
                    long delayTime = Long.parseLong(elements[1]);
                    if (delayTime == NOT_TIMESTAMP) {
                        queueData.setDelayTime(elements[1]);
                    }else {
                        Date tmpDate = new Date(delayTime);
                        queueData.setDelayTime(formatDate(tmpDate));
                    }
                }
                queueDataList.add(queueData);
            }

        }catch (Exception e){
            log.error(e.getMessage(),e);
        } finally {
            if (iterator != null) {
                iterator.close();
            }
        }
        return queueDataList;
    }

    private void output(QueueData queueData) {
        boolean flag = delayMessageOutListener.output(queueData);//消息输出监听器回调
        if (flag) {
            delete(queueData.getMsgId());
        }
    }

    public void compactRange() {
        try {
            db.compactRange();
        } catch (RocksDBException e) {
            log.error(e.getMessage(), e);
        }
    }

    @Override
    public void commit(String message) {
        this.insert(UUID.randomUUID().toString(), message, NOT_TIMESTAMP);
    }

    @Override
    public void commit(String msgId, String message) {
        this.insert(msgId, message, NOT_TIMESTAMP);
    }

    @Override
    public void commit(String message, Long timestamp) {
        this.insert(UUID.randomUUID().toString(), message, timestamp);
    }

    @Override
    public void commit(String msgId, String message, Long timestamp) {
        this.insert(msgId, message, timestamp);
    }

    private String formatDate(Date date, Object... pattern) {
        String formatDate;
        if (pattern != null && pattern.length > 0) {
            formatDate = DateFormatUtils.format(date, pattern[0].toString());
        } else {
            formatDate = DateFormatUtils.format(date, "yyyy-MM-dd HH:mm:ss");
        }
        return formatDate;
    }
}
